package com.xy.webadmin.app.service;

import com.xy.webadmin.app.constant.MqttConstants;
import com.xy.webadmin.app.model.MqttBrokerConfig;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * MQTT客户端管理服务 - 基于Spring Integration MQTT
 *
 * @author xiaoyue
 * @date 2025-07-03
 */
@Slf4j
@Service
public class MqttClientManager {
    // 使用两层Map: 外层按companyId分组，内层存储每个配置的MessageHandler实例
    private final Map<Long, Map<Long, MessageHandler>> mqttOutboundHandlerCache = new ConcurrentHashMap<>();
    
    // 缓存客户端工厂
    private final Map<Long, MqttPahoClientFactory> clientFactoryCache = new ConcurrentHashMap<>();
    
    // 缓存配置对象
    private final Map<Long, MqttBrokerConfig> configCache = new ConcurrentHashMap<>();
    
    @Autowired
    private DefaultPahoMessageConverter messageConverter;
    
    @Autowired
    private MqttPahoClientFactory defaultMqttClientFactory;
    
    /**
     * 获取MQTT消息处理器
     * 
     * @param companyId 公司ID
     * @param configId 配置ID
     * @return MQTT消息处理器
     */
    public MessageHandler getMqttOutboundHandler(Long companyId, Long configId) {
        return mqttOutboundHandlerCache.getOrDefault(companyId, new ConcurrentHashMap<>())
                .get(configId);
    }
    
    /**
     * 获取MQTT客户端工厂
     * 如果指定配置ID的工厂不存在，返回默认工厂
     * 
     * @param configId 配置ID
     * @return MQTT客户端工厂
     */
    public MqttPahoClientFactory getMqttClientFactory(Long configId) {
        MqttPahoClientFactory factory = clientFactoryCache.get(configId);
        if (factory == null) {
            log.warn("未找到配置ID为{}的MQTT客户端工厂，返回默认工厂", configId);
            return defaultMqttClientFactory;
        }
        return factory;
    }
    
    /**
     * 获取MQTT配置
     * 
     * @param configId 配置ID
     * @return MQTT配置对象
     */
    public MqttBrokerConfig getMqttConfig(Long configId) {
        return configCache.get(configId);
    }
    
    /**
     * 注册MQTT客户端
     * 
     * @param config MQTT配置对象
     */
    public synchronized void registerMqttClient(MqttBrokerConfig config) {
        try {
            // 检查配置是否有效
            if (config == null) {
                log.error("MQTT配置为null，无法注册客户端");
                return;
            }
            
            // 确保必要字段不为空
            if (config.getId() == null || config.getCompanyId() == null) {
                log.error("MQTT配置ID或公司ID为null，无法注册客户端");
                return;
            }
            
            // 检查broker URL是否存在
            if (config.getBrokerUrl() == null || config.getBrokerUrl().isEmpty()) {
                log.error("MQTT代理URL为空，无法注册客户端, configId: {}", config.getId());
                return;
            }
            
            log.info("开始为配置ID: {}创建MQTT客户端工厂, brokerUrl: {}", config.getId(), config.getBrokerUrl());
            
            // 创建MQTT客户端工厂
            MqttPahoClientFactory clientFactory = createMqttClientFactory(config);
            clientFactoryCache.put(config.getId(), clientFactory);
            log.debug("MQTT客户端工厂已缓存, configId: {}", config.getId());
            
            // 创建MQTT消息处理器
            log.info("开始为配置ID: {}创建MQTT消息处理器", config.getId());
            MessageHandler outboundHandler = createMqttOutboundHandler(config, clientFactory);
            
            // 按公司ID分组存储消息处理器
            mqttOutboundHandlerCache.computeIfAbsent(config.getCompanyId(), k -> new ConcurrentHashMap<>())
                          .put(config.getId(), outboundHandler);
            
            // 缓存配置
            configCache.put(config.getId(), config);
            
            log.info("成功注册MQTT客户端, configId: {}, companyId: {}, clientId: {}, brokerUrl: {}", 
                    config.getId(), config.getCompanyId(), config.getClientId(), config.getBrokerUrl());
        } catch (Exception e) {
            log.error("MQTT客户端注册失败, configId: {}, brokerUrl: {}, error: {}", 
                    config != null ? config.getId() : "unknown", 
                    config != null ? config.getBrokerUrl() : "unknown", 
                    e.getMessage(), e);
        }
    }
    
    /**
     * 创建MQTT客户端工厂
     * 
     * @param config MQTT配置对象
     * @return MQTT客户端工厂
     */
    private MqttPahoClientFactory createMqttClientFactory(MqttBrokerConfig config) {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        
        MqttConnectOptions options = new MqttConnectOptions();
        
        // 检查并设置服务器URL
        if (config.getBrokerUrl() != null && !config.getBrokerUrl().isEmpty()) {
            String brokerUrl = config.getBrokerUrl();
            
            // 如果URL没有协议前缀，添加默认的tcp://前缀
            if (!brokerUrl.startsWith(MqttConstants.Url.TCP_PREFIX) && !brokerUrl.startsWith(MqttConstants.Url.SSL_PREFIX)) {
                brokerUrl = MqttConstants.Url.TCP_PREFIX + brokerUrl;
                log.info("为MQTT代理URL添加默认协议: {}", brokerUrl);
            }
            
            // 如果URL没有端口，添加默认的1883端口（或SSL的8883端口）
            if (!brokerUrl.matches(".*:\\d+$")) {
                if (brokerUrl.startsWith(MqttConstants.Url.SSL_PREFIX)) {
                    brokerUrl = brokerUrl + ":" + MqttConstants.Url.DEFAULT_MQTT_SSL_PORT;
                } else {
                    brokerUrl = brokerUrl + ":" + MqttConstants.Url.DEFAULT_MQTT_PORT;
                }
                log.info("为MQTT代理URL添加默认端口: {}", brokerUrl);
            }
            
            log.info("设置MQTT代理URL: {}, configId: {}", brokerUrl, config.getId());
            options.setServerURIs(new String[]{brokerUrl});
        } else {
            log.error("MQTT代理URL为空，无法创建连接, configId: {}", config.getId());
            throw new IllegalArgumentException("MQTT代理URL不能为空");
        }
        
        // 设置清除会话，默认为true
        options.setCleanSession(config.getCleanSession() == null || config.getCleanSession() == 1);
        log.debug("设置清除会话: {}, configId: {}", 
                (config.getCleanSession() == null || config.getCleanSession() == 1), config.getId());
        
        // 设置用户名和密码，如果有的话
        if (config.getUsername() != null) {
            log.debug("设置MQTT认证用户名: {}, configId: {}", config.getUsername(), config.getId());
            options.setUserName(config.getUsername());
        }
        
        if (config.getPassword() != null) {
            log.debug("设置MQTT认证密码, configId: {}", config.getId());
            options.setPassword(config.getPassword().toCharArray());
        }
        
        // 设置心跳间隔，如果为null则使用默认值
        int keepAlive = config.getKeepalive() != null ? 
                config.getKeepalive() : MqttConstants.ConnectionDefaults.KEEP_ALIVE_INTERVAL;
        log.debug("设置心跳间隔: {}秒, configId: {}", keepAlive, config.getId());
        options.setKeepAliveInterval(keepAlive);
        
        // 设置连接超时，如果为null则使用默认值
        int connectionTimeout = config.getConnectionTimeout() != null ? 
                config.getConnectionTimeout() : MqttConstants.ConnectionDefaults.CONNECTION_TIMEOUT;
        log.debug("设置连接超时: {}秒, configId: {}", connectionTimeout, config.getId());
        options.setConnectionTimeout(connectionTimeout);
        
        // 设置最大并发消息数，如果为null则使用默认值
        int maxInflight = config.getMaxInflight() != null ? 
                config.getMaxInflight() : MqttConstants.ConnectionDefaults.MAX_INFLIGHT;
        log.debug("设置最大并发消息数: {}, configId: {}", maxInflight, config.getId());
        options.setMaxInflight(maxInflight);
        
        // 设置自动重连，默认为true
        boolean autoReconnect = config.getAutoReconnect() == null || config.getAutoReconnect() == 1;
        log.debug("设置自动重连: {}, configId: {}", autoReconnect, config.getId());
        options.setAutomaticReconnect(autoReconnect);
        
        // SSL配置
        if (config.getSslEnabled() != null && config.getSslEnabled() == 1) {
            // 配置SSL选项
            log.info("为MQTT客户端配置SSL, configId: {}", config.getId());
            // 这里需要根据具体SSL证书配置进行扩展
        }
        
        // 遗嘱消息
        if (config.getWillTopic() != null && !config.getWillTopic().isEmpty()) {
            String willPayload = config.getWillPayload() != null ? 
                    config.getWillPayload() : MqttConstants.ConnectionDefaults.DEFAULT_WILL_MESSAGE;
            
            int willQos = config.getWillQos() != null ? 
                    config.getWillQos() : MqttConstants.QosLevel.AT_LEAST_ONCE;
            
            boolean willRetain = config.getWillRetain() != null && config.getWillRetain() == 1;
            
            log.debug("设置遗嘱消息: topic={}, qos={}, retain={}, configId: {}", 
                    config.getWillTopic(), willQos, willRetain, config.getId());
            
            options.setWill(
                config.getWillTopic(),
                willPayload.getBytes(),
                willQos,
                willRetain
            );
        } else {
            // 设置默认遗嘱消息
            log.debug("设置默认遗嘱消息, configId: {}", config.getId());
            options.setWill(
                MqttConstants.ConnectionDefaults.DEFAULT_WILL_TOPIC,
                MqttConstants.ConnectionDefaults.DEFAULT_WILL_MESSAGE.getBytes(),
                MqttConstants.QosLevel.AT_LEAST_ONCE,
                false
            );
        }
        
        factory.setConnectionOptions(options);
        log.info("MQTT客户端工厂创建完成, configId: {}, brokerUrl: {}", config.getId(), config.getBrokerUrl());
        return factory;
    }
    
    /**
     * 创建MQTT输出处理器
     * 
     * @param config MQTT配置对象
     * @param clientFactory 客户端工厂
     * @return MQTT消息处理器
     */
    private MessageHandler createMqttOutboundHandler(MqttBrokerConfig config, MqttPahoClientFactory clientFactory) {
        // 确保客户端ID不为空
        String clientId = config.getClientId();
        if (clientId == null || clientId.isEmpty()) {
            clientId = MqttConstants.ConnectionDefaults.CLIENT_ID_PREFIX + config.getId() + "_" + System.currentTimeMillis();
            log.warn("MQTT客户端ID为空，使用生成的ID: {}", clientId);
        }
        
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, clientFactory);
        
        messageHandler.setAsync(true);
        
        // 设置QoS级别，如果为null则使用默认值
        int qosLevel = config.getQosLevel() != null ? 
                config.getQosLevel() : MqttConstants.QosLevel.DEFAULT;
        messageHandler.setDefaultQos(qosLevel);
        
        messageHandler.setConverter(messageConverter);
        
        // 初始化处理器
        messageHandler.setBeanName(MqttConstants.SpringIntegration.BEAN_NAME_OUTBOUND_HANDLER + config.getId());
        try {
            messageHandler.afterPropertiesSet();
        } catch (Exception e) {
            log.error("MQTT消息处理器初始化失败, configId: {}", config.getId(), e);
            throw new RuntimeException("MQTT消息处理器初始化失败", e);
        }
        
        return messageHandler;
    }
    
    /**
     * 获取所有MQTT配置
     * 
     * @return MQTT配置列表
     */
    public List<MqttBrokerConfig> getAllConfigs() {
        return new ArrayList<>(configCache.values());
    }
    
    /**
     * 断开所有MQTT客户端
     */
    @PreDestroy
    public void disconnectAll() {
        mqttOutboundHandlerCache.values().forEach(map -> 
            map.values().forEach(handler -> {
                try {
                    if (handler instanceof MqttPahoMessageHandler) {
                        ((MqttPahoMessageHandler) handler).stop();
                    }
                } catch (Exception e) {
                    log.error("断开MQTT连接失败", e);
                }
            })
        );
        log.info("已断开所有MQTT客户端连接");
    }
} 